package gspushmq.v1.websocket;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import gspushmq.model.ApiMessage;
import gspushmq.model.Message;
import gspushmq.model.MqType;
import gspushmq.model.User;
import gspushmq.util.GGLogger;
import gspushmq.util.ToolsUtil;
import gspushmq.v1.consumer.LoginUserThread;
import gspushmq.v1.subscribe.GGSubscribe;
import gspushmq.v1.subscribe.HandlerSubscribeImpl;
import gspushmq.v1.subscribe.send.core.GGMessageSend;
import gspushmq.v1.thread.GGThreadPool;
import gspushmq.v1.websocket.core.GGWebSocket;
import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ClientHandshake;

import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 *
 * 建立websocket、处理消息的入口
 * @version 1.0
 * @since JDK1.7
 * @author yutao
 * @company 上海朝阳永续信息技术有限公司
 * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
 * @date 2018年10月31日下午7:38:46
 */
public class GGWebSocketImpl extends GGWebSocket {

	public GGWebSocketImpl(int port) throws UnknownHostException{
		super( new InetSocketAddress(port));
	}
	
	//建立连接
	@Override
	public void onOpen(WebSocket wst, ClientHandshake chs) {
		GGLogger.record(new StringBuffer("action=建立连接")
				.append("|ip=").append(wst.getRemoteSocketAddress().getAddress().getHostAddress())
				.append("|websocket=").append(wst.toString())
				.append("|client_size=").append(this.getConnections().size()));
		String descriptor = chs.getResourceDescriptor();
		if(descriptor.contains("token")){
			int indexOf = descriptor.indexOf("?");
			String subStr = descriptor.substring(indexOf+1);
			Map<String, String> paramMap = ToolsUtil.getParamMap(subStr);
			String token = paramMap.get("token");
			Map<String, String> pMap = new HashMap<>();
			pMap.put("token", token);
			Map<String, String> clientFlag = new HashMap<>();
			try {
				JSONObject jsonObject = GGMessageSend.commonUrl(pMap, "v1/gguser_info/get");
				if(jsonObject == null || !jsonObject.containsKey("data")){
					clientFlag.put("code", "1001");
					wst.send(JSONObject.toJSONString(clientFlag));
					GGLogger.error(getClass(), "token失效");
					return;
				}
				JSONObject dataJson = jsonObject.getJSONObject("data");
				Integer orgId = dataJson.getInteger("org_id");
				String aidLong = dataJson.getString("account_id");
				String accountName = dataJson.getString("account");
				User user = new User();
				user.setAccountId(aidLong);
				user.setOrgId(orgId);
				user.setAccountName(accountName);
				String uuid = UUID.randomUUID().toString().replaceAll("-", "");
				user.setUuid(uuid);
				//===========获取权限并设置============
				String productLine = paramMap.get("product_line");
				user.setProductLine(productLine);
				
				getPermi(Long.valueOf(aidLong), productLine, user);
				
				GGSubscribe.wstUser.put(wst, user);
				
//				Map<String, String> clientFlag = new HashMap<>();
				clientFlag.clear();
				clientFlag.put("client_flag", uuid);
				wst.send(JSONObject.toJSONString(clientFlag));
				
				writeLog(wst, "onOpen");
				
				//开始上线消费线程，加入超时处理
				GGThreadPool.threadPool.commitTask(new LoginUserThread(user,wst));
			} catch (Exception e) {
				wst.send(e.toString());
				GGSubscribe.wstUser.remove(wst);
				wst.close();
				GGLogger.error(getClass(), "建立websocket时，非法请求"+e.getMessage());
			}
		}else{
			//非法连接，关闭websocket
			GGSubscribe.wstUser.remove(wst);
			wst.close();
		}
	}
	
	
	/**
	 * 获取用户终端产品
	 * @param accountId
	 * @param productLine
	 * @author yutao
	 * @date 2018年11月13日下午5:43:56
	 */
	private void getPermi(Long accountId, String productLine, User user) throws Exception{
		Map<String, String> pMap = new HashMap<>();
		pMap.put("source", productLine);
		pMap.put("is_activate", "1");
		pMap.put("account_id", accountId.toString());
		
		JSONObject perm = GGMessageSend.commonUrl(pMap, "v1/user/get_permission_data");
		if(perm == null){
			throw new Exception("获取用户权限失败");
		}
		JSONObject permData = perm.getJSONObject("data");
		if(permData == null){
			throw new Exception("用户权限解析失败");
		}
		String permi = permData.getString("permi");
		Integer userType = permData.getInteger("user_type");
		user.setPermi(permi);
		user.setProType(userType);
	}

	@Override
	public void onClose(WebSocket wst, int code, String reason, boolean remote) {
		
		GGLogger.info(this.getClass(),"Pusher onClose code:" + code + ",reason: " + reason + ", remote: " + remote);
		writeLog(wst, "onClose");
		cleanMemory(wst);
	}

	/**
	 * {"requestId":"ai", "productLine":"901","accountId":"347209","subStatus":"1", "clientFlag":"###","mqType":"api","mqSubType":"count","proType":"1","params":{'get_count':'1','type':'0'}, apiUrl:"v1/treasure/list_clue"}
	 * 
	 */
	@Override
	public void onMessage(WebSocket wst, String message) {
		//处理具体的业务
		if(!wst.isOpen()){
			GGLogger.info(this.getClass(),"websocket状态不是打开的");
			return;
		}
		GGLogger.record(new StringBuffer("action=接收消息")
						.append("|ip=").append(wst.getRemoteSocketAddress().getAddress().getHostAddress())
						.append("|websocket=").append(wst.toString())
						.append("|").append(message.replace(":", "=")));
		
		GGLogger.info(this.getClass(),"Pusher onMessage(String): " + message);
		try {
			//解析字符串 变成 Message 对象
			Message mess = JSON.parseObject(message, Message.class);
			
			String subStatus = mess.getSub_status();
			if("2".equals(subStatus)) {
				String aidLong = mess.getAccount_id();
				String productLine = mess.getProduct_line();
				User user = GGSubscribe.wstUser.get(wst);
				if(user == null) {
					return;
				}
				getPermi(Long.valueOf(aidLong), productLine, user);
				
				writeLog(wst, "onMessage--subStatus--2", mess);
				
				Map<String, Object> map = new HashMap<>();
				map.put("request_id", mess.getRequest_id());
				wst.send(JSONObject.toJSONString(map));
				return;
			}
			
			String mqType = mess.getMq_type();
			switch (mqType) {
			case MqType.API:
				mess = JSON.parseObject(message, ApiMessage.class);
				break;
			default:
				break;
			}
			GGSubscribe.wstMessage.put(wst, mess);

			GGThreadPool.threadPool.commit(new HandlerSubscribeImpl(mess, wst));
			GGLogger.warn(getClass(), "--onMessage--commit");
			writeLog(wst, "onMessage", mess);
		} catch (Exception e) {
			// TODO: handle exception
			GGLogger.error(getClass(),"onMessage-Error:"+e);
		}
		
	}

	@Override
	public void onError(WebSocket wst, Exception e) {
		
		GGLogger.error(this.getClass(), e);
		if( wst != null ) {
			GGLogger.record(e, new StringBuffer("action=异常信息")
					.append("|ip=").append(wst.getRemoteSocketAddress().getAddress().getHostAddress())
					.append("|websocket=").append(wst.toString())
					.append("|content="));
		
		}

		writeLog(wst, "onError");
		cleanMemory(wst);
		if (wst != null) {
			wst.close(1006, "onError");
		}
	}

	/**
	 * 
	 * @see org.java_websocket.server.WebSocketServer#onStart()
	 */
	@Override
	public void onStart() {
		GGLogger.info(getClass(), "---websocket---启动成功---");
	}

}